package com.atguigu.actual.edu0417.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.beans.CartAddUuBean;
import com.atguigu.actual.edu0417.beans.DwsInteractionReview;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.func.BeanToJsonStrMapFcuntion;
import com.atguigu.actual.edu0417.func.DimAsyncFunction;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.DorisSinkUtil;
import com.atguigu.actual.edu0417.unils.DorisUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Locale;
import java.util.concurrent.TimeUnit;

/**
 * @author: 洛尘
 * @since: 2023-10-22 11:40
 * @description:
 * 课程评价
 * 度量值  评价几个星，好评用户数
 **/
public class DwsInteractionReviewWindow extends BaseApp {
    //基本环境准备
    public static void main(String[] args) {
        DwsInteractionReviewWindow dwsTradeCourseOrderWindow = new DwsInteractionReviewWindow();
        dwsTradeCourseOrderWindow.base(args);
    }
    //业务逻辑处理
    //读取主题是dwdBaseApp中的dwd_interaction_review_add
    @Override
    public void process(StreamExecutionEnvironment env) {
        //创建消费者
        String groupID="dws_interaction_review_add_window";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(EduConfig.KAFKA_DWD_INTERACTION_REVIEW_ADD_DB, groupID);
        //封装成流
        SingleOutputStreamOperator<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "review_kafkasource");
//        kafkaDS.print("*******");
        //类型转换
        SingleOutputStreamOperator<DwsInteractionReview> beanDS = kafkaDS.map(
                new MapFunction<String, DwsInteractionReview>() {
                    @Override
                    public DwsInteractionReview map(String jsonStr) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                        String courseId = jsonObject.getString("course_id");
                        Long reviewStars = jsonObject.getLong("review_stars");
                        Long ts = jsonObject.getLong("ts") * 1000;
                        DwsInteractionReview eduBean = DwsInteractionReview.builder()
                                .courseId(courseId)
                                .reviewTotalStarts(reviewStars)
                                .reviewUserCount(1L)
                                .goodReviewUserCount(reviewStars == 5 ? 1L : 0L)
                                .ts(ts)
                                .build();
                        return eduBean;
                    }
                }
        );
//        beanDS.print("bean>>>");
        //添加水位线
        SingleOutputStreamOperator<DwsInteractionReview> waterDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsInteractionReview>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsInteractionReview>() {
                                    @Override
                                    public long extractTimestamp(DwsInteractionReview element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );
        //按照课程id分组
        KeyedStream<DwsInteractionReview, String> keyedDS = waterDS.keyBy(DwsInteractionReview::getCourseId);
//        keyedDS.print("key>>>");
        //开窗聚合，补充字段
        WindowedStream<DwsInteractionReview, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        SingleOutputStreamOperator<DwsInteractionReview> reduceDS = windowDS.reduce(
                new ReduceFunction<DwsInteractionReview>() {
                    @Override
                    public DwsInteractionReview reduce(DwsInteractionReview value1, DwsInteractionReview value2) throws Exception {
                        value1.setReviewUserCount(value1.getReviewUserCount() + value2.getReviewUserCount());
                        value1.setGoodReviewUserCount(value1.getReviewUserCount() + value2.getGoodReviewUserCount());
                        value1.setReviewTotalStarts(value1.getReviewTotalStarts() + value2.getReviewTotalStarts());
                        return value1;
                    }
                },
                new ProcessWindowFunction<DwsInteractionReview, DwsInteractionReview, String, TimeWindow>() {
                    @Override
                    public void process(String jsonStr, Context context, Iterable<DwsInteractionReview> elements, Collector<DwsInteractionReview> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String curDate = DateFormatUtil.toDate(context.window().getStart());
                        for (DwsInteractionReview dwsInteractionReview : elements) {
                            dwsInteractionReview.setStt(stt);
                            dwsInteractionReview.setEdt(edt);
                            dwsInteractionReview.setCurDate(curDate);
                            out.collect(dwsInteractionReview);
                        }
                    }
                }
        );
        reduceDS.print(">>>???");
//        补充字段维度关联
        SingleOutputStreamOperator<DwsInteractionReview> dimDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<DwsInteractionReview>("dim_course_info") {
                    @Override
                    public void join(DwsInteractionReview bean, JSONObject dimCourseJsonObj) {
                        bean.setCourseName(dimCourseJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getKey(DwsInteractionReview bean) {
                        return bean.getCourseId();
                    }
                }, 5 * 60, TimeUnit.SECONDS
        );
        dimDS.print("dim>>>>");
        dimDS.addSink(DorisSinkUtil.<DwsInteractionReview>getJdbcSink("insert into dws_interaction_review_window values(?,?,?,?,?,?,?,?)"));
//        dimDS
//                .map(new BeanToJsonStrMapFcuntion<>())
//                .sinkTo(DorisUtil.getDorisSink("dws_interaction_review_window"));


    }
}